/*
package com.kyrie.controller;

import com.alibaba.fastjson.JSON;
import com.kyrie.annotation.TransLog;
import com.kyrie.dto.KafkaProducerDto;
import com.kyrie.system.mybatis.execption.ServiceException;
import com.kyrie.utils.GlobalConstants;
import com.kyrie.vo.ResponseBean;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.shiro.authz.annotation.Logical;
import org.apache.shiro.authz.annotation.RequiresRoles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import javax.validation.Valid;
import java.util.List;
import java.util.Optional;

*/
/**
 * 描述：kafka生产者controller
 * @author wuxiang
 * @date 2020-04-19 19:38:00
 *//*

@Validated
@RestController
public class KafkaProducerController {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @TransLog(transCode = "/produceMsg",description = "@kafka消息发布接口")
    @RequestMapping(value = "/produceMsg",method = RequestMethod.POST)
    @RequiresRoles(logical = Logical.OR, value = {"user","systemAdmin"})
    public ResponseBean<String> getLog(@Valid @RequestBody KafkaProducerDto kafkaProducerDto) throws ServiceException {
        kafkaTemplate.send("topic-kafka-demo", JSON.toJSONString(kafkaProducerDto.getPushContent()));

        return ResponseBean.success(null, GlobalConstants.SUCCESS_CODE,"@kafka消息发布接口");
    }

    @KafkaListener(topics = { "topic-kafka-demo" })
    public void receive(ConsumerRecord<?, ?> record) {

        logger.info("消费得到的消息---key: " + record.key());
        logger.info("消费得到的消息---value: " + record.value().toString());

    }

    */
/**
     * 批量消费自动提交
     *//*

    @KafkaListener(topics ="batchAuto",containerFactory="kafkaListenerContainerFactory")
    public void batchReceive(List<ConsumerRecord<?, ?>> records) {
        logger.info("接收到消息数量：{}",records.size());
        for(ConsumerRecord record: records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            logger.info("Received:{} ",record);
            if (kafkaMessage.isPresent()) {
                Object message = record.value();
                String topic = record.topic();
                logger.info("消费得到的消息:{},topic:{} ",message,topic);
            }
        }

    }

    */
/**
     * 批量消费手动提交
     *//*

    @KafkaListener(topics = "batchManual", containerFactory="kafkaListenerContainerFactory")
    public void batchReceive(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment){
        logger.info("接收到消息数量：{}",records.size());
        for(ConsumerRecord record: records) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            logger.info("Received: {}", record);
            if (kafkaMessage.isPresent()) {
                Object message = record.value();
                String topic = record.topic();
                logger.info("消费得到的消息:{},topic:{} ",message,topic);
            }
        }
        acknowledgment.acknowledge();
    }
}
*/
